package com.wl.cloud.datax.common.utils;

import cn.hutool.core.util.ObjUtil;
import com.alibaba.druid.DbType;
import com.alibaba.druid.util.MySqlUtils;
import com.alibaba.druid.util.OracleUtils;
import com.alibaba.druid.util.PGUtils;
import com.alibaba.druid.util.ResultSetConsumer;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.wl.cloud.common.model.DatasourceModel;
import com.wl.cloud.core.exception.CustomException;
import com.wl.cloud.datax.common.enums.DBUtilErrorCode;
import com.wl.cloud.datax.common.exception.RdbmsException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;

import java.sql.*;
import java.util.*;

import static com.wl.cloud.datax.common.utils.DataBaseType.*;

/**
 * @author: wanglin
 * @date: 2023-07-12 周三
 * @Version: 1.0
 * @Description:
 */
@Slf4j
public class JdbcCustomUtils {

    public static void close(ResultSet rs, Statement stat, Connection conn) {
        close(rs);
        close(stat);
        close(conn);
    }

    public static void close(@Nullable Connection x) {
        if (x == null) {
            return;
        }

        try {
            if (x.isClosed()) {
                return;
            }

            x.close();
        } catch (SQLRecoverableException e) {
            // skip
        } catch (Exception e) {
            log.debug("close connection error", e);
        }
    }

    public static void close(Statement x) {
        if (x == null) {
            return;
        }
        try {
            x.close();
        } catch (Exception e) {
            boolean printError = true;

            if (e instanceof java.sql.SQLRecoverableException
                    && "Closed Connection".equals(e.getMessage())
            ) {
                printError = false;
            }

            if (printError) {
                log.debug("close statement error", e);
            }
        }
    }

    public static void close(ResultSet x) {
        if (x == null) {
            return;
        }
        try {
            x.close();
        } catch (Exception e) {
            log.debug("close result set error", e);
        }
    }

    /**
     * 判断数据源类型，mysql，hive，加载名称，默认是自动加载的，不需要收到加载
     *
     * @param dataBaseType
     * @return
     */
    public static String getDriverClassName(DataBaseType dataBaseType) {
        if (ObjUtil.isNull(dataBaseType)) {
            return null;
        }
        String driverName;
        switch (dataBaseType) {
            case Hive:
                driverName = Hive.getValue();
                break;
            case MySql:
                driverName = MySql.getValue();
                break;
            case Oracle:
                driverName = Oracle.getValue();
                break;
            default:
                throw new CustomException("暂不支持所选数据库类型：" + dataBaseType.getKey());
        }
        return driverName;
    }

    public static Driver createDriver(String driverClassName) throws SQLException {
        return createDriver(null, driverClassName);
    }

    public static Driver createDriver(ClassLoader classLoader, String driverClassName) throws SQLException {
        Class<?> clazz = null;
        if (classLoader != null) {
            try {
                clazz = classLoader.loadClass(driverClassName);
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
                log.error(String.format("数据库: %s 驱动加载失败!", driverClassName));
                throw CustomException.asCustomException(String.format("数据库: %s 驱动加载失败!", driverClassName));
            }
        }

        if (clazz == null) {
            try {
                ClassLoader contextLoader = Thread.currentThread().getContextClassLoader();
                if (contextLoader != null) {
                    clazz = contextLoader.loadClass(driverClassName);
                }
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
                log.error(String.format("数据库: %s 驱动加载失败!", driverClassName));
                throw CustomException.asCustomException(String.format("数据库: %s 驱动加载失败!", driverClassName));
            }
        }

        if (clazz == null) {
            try {
                clazz = Class.forName(driverClassName);
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
                log.error(String.format("数据库: %s 驱动加载失败!", driverClassName));
                throw CustomException.asCustomException(String.format("数据库: %s 驱动加载失败!", driverClassName));
            }
        }

        try {
            return (Driver) clazz.newInstance();
        } catch (Exception e) {
            e.printStackTrace();
            log.error(String.format("数据库: %s 驱动加载失败!", driverClassName));
            throw CustomException.asCustomException(String.format("数据库: %s 驱动加载失败!", driverClassName));
        }
    }

    public static void execute(DatasourceModel dataSource, String sql, Object... parameters) throws SQLException {
        execute(dataSource, sql, Arrays.asList(parameters));
    }

    public static void execute(DatasourceModel dataSource, String sql, List<Object> parameters) throws SQLException {
        Connection conn = null;
        try {
            conn = dataSource.getConnection();
            execute(conn, sql, parameters);
        } finally {
            close(conn);
        }
    }

    public static void execute(Connection conn, String sql) throws SQLException {
        execute(conn, sql, Collections.emptyList());
    }

    public static void execute(Connection conn, String sql, List<Object> parameters) throws SQLException {
        PreparedStatement stmt = null;

        try {
            stmt = conn.prepareStatement(sql);

            setParameters(stmt, parameters);

            stmt.executeUpdate();
        } finally {
            JdbcCustomUtils.close(stmt);
        }
    }

    public static Connection getConnection(DatasourceModel datasourceModel) {
        try {
            createDriver(getEnumByKey(datasourceModel.getType()).getValue());
            return datasourceModel.getConnection();
        } catch (SQLException throwables) {
            throwables.printStackTrace();
            log.error("数据库: " + datasourceModel.getType() + "连接失败，请检查连接信息是否正确！");
            throw CustomException.asCustomException("数据库: " + datasourceModel.getType() + "连接失败，请检查连接信息是否正确！");
        }
    }

    public static Connection getConnection(DataBaseType dataBaseType, DatasourceModel datasourceModel) {
        try {
            createDriver(dataBaseType.getValue());
            return datasourceModel.getConnection();
        } catch (SQLException throwables) {
            throwables.printStackTrace();
            log.error("数据库: " + dataBaseType.getKey() + "连接失败，请检查连接信息是否正确！");
            throw CustomException.asCustomException("数据库: " + dataBaseType.getKey() + "连接失败，请检查连接信息是否正确！");
        }
    }

    public static Connection getConnection(DatasourceModel datasourceModel, String dbName) {
        DataBaseType dataBaseType = getEnumByKey(datasourceModel.getType());
        Assert.notNull(dataBaseType, "DataX暂不支持改该数据库类型: " + datasourceModel.getType());
        try {
            createDriver(dataBaseType.getValue());
            if (StringUtils.isNotBlank(dbName)) {
                datasourceModel.setUrl(dataBaseType.appendJDBCSuffixForReader(datasourceModel.getUrl(), dbName));
            }
            return datasourceModel.getConnection();
        } catch (SQLException throwables) {
            throwables.printStackTrace();
            log.error("数据库: " + datasourceModel.getType() + "连接失败，数据库名：" + dbName + ",请检查连接信息是否正确！");
            throw CustomException.asCustomException("数据库: " + datasourceModel.getType() + "连接失败，数据库名：" + dbName + ",请检查连接信息是否正确！");
        }
    }


//    protected java.sql.Connection getConnection(DatasourceModel datasourceModel) throws SQLException {
//        Assert.notNull(datasourceModel, "datasource error is null");
//        Assert.isTrue(StringUtils.isNotBlank(datasourceModel.getUrl()), "数据库连接信息不能为空");
//        Assert.isTrue(StringUtils.isNotBlank(datasourceModel.getUsername()), "数据库用户名不能为空");
//        Assert.isTrue(StringUtils.isNotBlank(datasourceModel.getPassword()), "数据库密码不能为空");
//        return DriverManager.getConnection(datasourceModel.getUrl(), datasourceModel.getUsername(), datasourceModel.getPassword());
//    }

    protected java.sql.Connection getConnection(String url, String username, String password) throws SQLException {
        return DriverManager.getConnection(url, username, password);
    }

    private static void setParameters(PreparedStatement stmt, List<Object> parameters) throws SQLException {
        for (int i = 0, size = parameters.size(); i < size; ++i) {
            Object param = parameters.get(i);
            stmt.setObject(i + 1, param);
        }
    }

    /**
     * SQL查询
     *
     * @param dataSource
     * @param sql
     * @param parameters
     * @return
     * @throws SQLException
     */
    public static List<Map<String, Object>> executeQuery(DatasourceModel dataSource, String sql, Object... parameters) throws SQLException {
        return executeQuery(dataSource, sql, Arrays.asList(parameters));
    }

    public static List<Map<String, Object>> executeQuery(DatasourceModel dataSource, String sql, List<Object> parameters) throws SQLException {
        Connection conn = null;
        try {
            conn = dataSource.getConnection();
            return executeQuery(conn, sql, parameters);
        } finally {
            close(conn);
        }
    }

    public static List<Map<String, Object>> executeQuery(Connection conn, String sql, List<Object> parameters) throws SQLException {
        List<Map<String, Object>> rows = new ArrayList<>();

        PreparedStatement stmt = null;
        ResultSet rs = null;
        try {
            stmt = conn.prepareStatement(sql);
            setParameters(stmt, parameters);
            rs = stmt.executeQuery();
            ResultSetMetaData rsMeta = rs.getMetaData();

            while (rs.next()) {
                Map<String, Object> row = new LinkedHashMap<>();
                for (int i = 0, size = rsMeta.getColumnCount(); i < size; ++i) {
                    String columnName = rsMeta.getColumnLabel(i + 1);
                    Object value = rs.getObject(i + 1);
                    row.put(columnName, value);
                }
                rows.add(row);
            }
        } finally {
            JdbcCustomUtils.close(rs);
            JdbcCustomUtils.close(stmt);
        }

        return rows;
    }


    /**
     * 返回查询结果
     *
     * @param dataSource
     * @param sql
     * @return
     * @throws SQLException
     */
    public static List<String> executeQuery(DatasourceModel dataSource, String sql) throws SQLException {
        Connection conn = null;
        PreparedStatement stmt = null;
        ResultSet rs = null;
        List<String> resultList = new ArrayList<>();
        try {
            conn = dataSource.getConnection();
            stmt = conn.prepareStatement(sql);
            rs = stmt.executeQuery();
            ResultSetMetaData rsMeta = rs.getMetaData();
            while (rs.next()) {
                for (int i = 0, size = rsMeta.getColumnCount(); i < size; ++i) {
                    resultList.add(rs.getString(i + 1));
                }
            }
        } finally {
            close(rs);
            close(stmt);
            close(conn);
        }
        return resultList;
    }


    public static <T> void executeQuery(DatasourceModel dataSource, ResultSetConsumer<T> consumer, String sql, Object... parameters) throws SQLException {
        Connection conn = null;
        PreparedStatement stmt = null;
        ResultSet rs = null;
        try {
            conn = dataSource.getConnection();
            stmt = conn.prepareStatement(sql);
            for (int i = 0; i < parameters.length; ++i) {
                stmt.setObject(i + 1, parameters[i]);
            }
            rs = stmt.executeQuery();
            while (rs.next()) {
                if (consumer != null) {
                    T object = consumer.apply(rs);
                    consumer.accept(object);
                }
            }
        } finally {
            close(rs);
            close(stmt);
            close(conn);
        }
    }

    /**
     * 批量插入
     *
     * @param dataSource
     * @param tableName
     * @param data
     * @throws SQLException
     */
    public static void insertToTable(DatasourceModel dataSource, String tableName, Map<String, Object> data)
            throws SQLException {
        Connection conn = null;
        try {
            conn = dataSource.getConnection();
            insertToTable(conn, tableName, data);
        } finally {
            close(conn);
        }
    }

    public static void insertToTable(Connection conn, String tableName, Map<String, Object> data) throws SQLException {
        String sql = makeInsertToTableSql(tableName, data.keySet());
        List<Object> parameters = new ArrayList<Object>(data.values());
        execute(conn, sql, parameters);
    }

    public static String makeInsertToTableSql(String tableName, Collection<String> names) {
        StringBuilder sql = new StringBuilder() //
                .append("insert into ") //
                .append(tableName) //
                .append("("); //

        int nameCount = 0;
        for (String name : names) {
            if (nameCount > 0) {
                sql.append(",");
            }
            sql.append(name);
            nameCount++;
        }
        sql.append(") values (");
        for (int i = 0; i < nameCount; ++i) {
            if (i != 0) {
                sql.append(",");
            }
            sql.append("?");
        }
        sql.append(")");

        return sql.toString();
    }

    /**
     * 验证SQL知否能执行成功（验证连接信息），不返回
     *
     * @param dataSource
     * @param sql
     * @throws SQLException
     */
    public static void executeQueryValidate(DatasourceModel dataSource, String sql) throws SQLException {
        Connection conn = null;
        PreparedStatement stmt = null;
        ResultSet rs = null;
        try {
            conn = dataSource.getConnection();
            stmt = conn.prepareStatement(sql);
            rs = stmt.executeQuery();
        } finally {
            close(rs);
            close(stmt);
            close(conn);
        }
    }

    public static List<String> showTables(Connection conn, DbType dbType) throws SQLException {
        if (DbType.mysql == dbType || DbType.oceanbase == dbType) {
            return MySqlUtils.showTables(conn);
        }

        if (dbType == DbType.oracle || dbType == DbType.oceanbase_oracle) {
            return OracleUtils.showTables(conn);
        }

        if (dbType == DbType.postgresql) {
            return PGUtils.showTables(conn);
        }
        throw new SQLException("show tables dbType not support for " + dbType);
    }

    public static List<String> showTables(Connection conn) throws SQLException {
        List<String> tables = new ArrayList<>();

        Statement stmt = null;
        ResultSet rs = null;
        try {
            stmt = conn.createStatement();
            rs = stmt.executeQuery("show tables");
            while (rs.next()) {
                String tableName = rs.getString(1);
                tables.add(tableName);
            }
        } finally {
            close(rs);
            close(stmt);
            close(conn);
        }
        return tables;
    }

    public static List<String> showDatabases(Connection conn) throws SQLException {
        List<String> databases = new ArrayList<>();

        Statement stmt = null;
        ResultSet rs = null;
        try {
            stmt = conn.createStatement();
            rs = stmt.executeQuery("show databases");
            while (rs.next()) {
                String databaseName = rs.getString(1);
                databases.add(databaseName);
            }
        } finally {
            close(rs);
            close(stmt);
            close(conn);
        }
        return databases;
    }

    public static List<String> showDatabases(DatasourceModel datasourceModel) {
        DataBaseType dataBaseType = getEnumByKey(datasourceModel.getType());
        Assert.notNull(dataBaseType, "DataX暂不支持改该数据库类型: " + datasourceModel.getType());

        Connection conn = getConnection(dataBaseType, datasourceModel);
        Statement stmt = null;
        ResultSet rs = null;
        String sql = createSelectDatabaseSql(dataBaseType);

        List<String> databases = new ArrayList<>();
        try {
            stmt = conn.createStatement();
            rs = stmt.executeQuery(sql);
            while (rs.next()) {
                for (int i = 1; i <= rs.getMetaData().getColumnCount(); i++) {
                    String databaseName = rs.getString(i);
                    databases.add(databaseName);
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            log.error("数据库查询失败，SQL语句：" + sql + "执行失败!");
            throw CustomException.asCustomException("数据库查询失败，SQL语句：" + sql + "执行失败!");
        } finally {
            close(rs);
            close(stmt);
            close(conn);
        }
        return databases;
    }

    private static String createSelectDatabaseSql(DataBaseType dataBaseType) {
        String sql = "";
        switch (dataBaseType) {
            case Hive:
            case MySql:
                sql = "show databases";
                break;
            case Oracle:
                sql = "select name from v$database";
                break;
            case SQLServer:
                sql = "select name from sys.databases";
                break;
            case PostgreSQL:
                sql = "select datname from pg_database";
                break;
            default:
                throw CustomException.asCustomException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type.");
        }
        return sql;
    }

    public static List<String> showTables(DatasourceModel datasourceModel, String dbName) {
        DataBaseType dataBaseType = getEnumByKey(datasourceModel.getType());
        Assert.notNull(dataBaseType, "DataX暂不支持改该数据库类型: " + datasourceModel.getType());

        Connection conn = getConnection(datasourceModel, dbName);
        Statement stmt = null;
        ResultSet rs = null;
        if (Oracle.equals(dataBaseType)) {
            dbName = datasourceModel.getUsername().toUpperCase();
        }
        String sql = createSelectTableSql(dataBaseType, dbName);

        List<String> tableNames = new ArrayList<>();
        try {
            stmt = conn.createStatement();
            rs = stmt.executeQuery(sql);
            while (rs.next()) {
                for (int i = 1; i <= rs.getMetaData().getColumnCount(); i++) {
                    tableNames.add(rs.getString(i));
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            log.error("数据库表查询失败，SQL语句：" + sql + "执行失败!");
            throw CustomException.asCustomException("数据库表查询失败，SQL语句：" + sql + "执行失败!");
        } finally {
            close(rs);
            close(stmt);
            close(conn);
        }
        return tableNames;
    }


    private static String createSelectTableSql(DataBaseType dataBaseType, String dbName) {
        String sql = "";
        switch (dataBaseType) {
            case Hive:
            case MySql:
                sql = "show tables";
                break;
            case Oracle:
                sql = "select table_name from all_tables where owner = '%s'";
                break;
            case SQLServer:
                sql = "select name from sys.tables";
                break;
            case PostgreSQL:
                sql = "select tablename from pg_tables where schemaname='public'";
                break;
            default:
                throw CustomException.asCustomException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type.");
        }
        return String.format(sql, dbName);
    }

    private static String createSelectTableFieldSql(DataBaseType dataBaseType, String dbName, String tbName) {
        String sql = "";
        switch (dataBaseType) {
            case Hive:
                sql = "describe " + dbName + "." + tbName;
                break;
            case MySql:
                sql = "SELECT COLUMN_NAME from information_schema.COLUMNS where table_schema = '" + dbName + "' and table_name = '" + tbName + "'";
                break;
            default:
                throw CustomException.asCustomException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type.");
        }
        return sql;
    }

    private static String createSelectTableFieldWithTypeSql(DataBaseType dataBaseType, String dbName, String tbName) {
        String sql = "";
        switch (dataBaseType) {
            case Hive:
                sql = "describe " + dbName + "." + tbName;
                break;
            case MySql:
                sql = "SELECT COLUMN_NAME,DATA_TYPE from information_schema.COLUMNS where table_schema = '" + dbName + "' and table_name = '" + tbName + "'";
                break;
            default:
                throw CustomException.asCustomException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type.");
        }
        return sql;
    }

//    public static List<String> dropTables(DatasourceModel datasourceModel, List<String> tableNameList) {
//        DataBaseType dataBaseType = DataBaseType.getEnumByKey(datasourceModel.getType());
//        Assert.notNull(dataBaseType, "DataX暂不支持改该数据库类型: " + datasourceModel.getType());
//
//        Connection conn = getConnection(dataBaseType, datasourceModel);
//        Statement stmt = null;
//        ResultSet rs = null;
//
//
//        try {
//            stmt = conn.createStatement();
//            for (int i = 0; i < tableNameList.size(); i++) {
//                stmt.execute("drop table if exists " + tableNameList.get(i));
//            }
//
//
//        } catch (SQLException e) {
//            e.printStackTrace();
//            log.error("数据库查询失败，SQL语句：" + sql + "执行失败!");
//            throw CustomException.asCustomException("数据库查询失败，SQL语句：" + sql + "执行失败!");
//        } finally {
//            close(rs);
//            close(stmt);
//            close(conn);
//        }
//        return databases;
//    }

    public static List<String> selectTableColumns(Connection conn, String tbName) throws SQLException {
        List<String> columnList = Lists.newArrayList();
        ResultSet rs = null;
        PreparedStatement stmt = null;
        try {
            stmt = conn.prepareStatement("select * from " + tbName + " where 1=2");
            rs = stmt.executeQuery();
            ResultSetMetaData metaData = rs.getMetaData();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                columnList.add(metaData.getColumnName(i));
            }
        } finally {
            close(rs, stmt, null);
        }
        return columnList;
    }


    public static List<String> showTableFields(DatasourceModel datasourceModel, String dbName, String tbName) {
        DataBaseType dataBaseType = getEnumByKey(datasourceModel.getType());
        Assert.notNull(dataBaseType, "DataX暂不支持改该数据库类型: " + datasourceModel.getType());

        Connection conn = getConnection(datasourceModel);
        Statement stmt = null;
        ResultSet rs = null;
        if (Oracle.equals(dataBaseType)) {
            dbName = datasourceModel.getUsername().toUpperCase();
        }
        String sql = createSelectTableFieldSql(dataBaseType, dbName, tbName);

        List<String> fieldNames = new ArrayList<>();
        try {
            stmt = conn.createStatement();
            rs = stmt.executeQuery(sql);
            String dbColName = "";
            switch (dataBaseType) {
                case Hive:
                    dbColName = "col_name";
                    break;
                case MySql:
                    dbColName = "COLUMN_NAME";
                    break;
                default:
                    throw CustomException.asCustomException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type.");
            }
            while (rs.next()) {
                String fieldName = rs.getString(dbColName);
                if (StringUtils.isNotBlank(fieldName)) {
                    fieldNames.add(fieldName);
                } else {
                    break;
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            log.error("SQL语句：" + sql + "执行失败!");
            throw CustomException.asCustomException("SQL语句：" + sql + "执行失败!");
        } finally {
            close(rs);
            close(stmt);
            close(conn);
        }
        return fieldNames;
    }

    public static List<Map<String, String>> showTableFieldsWithType(DatasourceModel datasourceModel, String dbName, String tbName) {
        DataBaseType dataBaseType = getEnumByKey(datasourceModel.getType());
        Assert.notNull(dataBaseType, "DataX暂不支持改该数据库类型: " + datasourceModel.getType());

        Connection conn = getConnection(datasourceModel);
        Statement stmt = null;
        ResultSet rs = null;
        if (Oracle.equals(dataBaseType)) {
            dbName = datasourceModel.getUsername().toUpperCase();
        }
        String sql = createSelectTableFieldWithTypeSql(dataBaseType, dbName, tbName);

        List<Map<String, String>> fieldNamesWithType = new ArrayList<>();
        try {
            stmt = conn.createStatement();
            rs = stmt.executeQuery(sql);
            String dbColName = "";
            String dbDataType = "";
            switch (dataBaseType) {
                case Hive:
                    dbColName = "col_name";
                    dbDataType = "data_type";
                    break;
                case MySql:
                    dbColName = "COLUMN_NAME";
                    dbDataType = "DATA_TYPE";
                    break;
                default:
                    throw CustomException.asCustomException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type.");
            }
            while (rs.next()) {
                String fieldName = rs.getString(dbColName);
                String dataType = rs.getString(dbDataType);
                if (StringUtils.isNotBlank(fieldName) && StringUtils.isNotBlank(dataType)) {
                    Map<String, String> map = new HashMap<>();
                    map.put("colName", fieldName);
                    map.put("dataType", dataType);
                    fieldNamesWithType.add(map);
                } else {
                    break;
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            log.info("SQL语句：" + sql + "执行失败!");
            throw CustomException.asCustomException("SQL语句：" + sql + "执行失败!");
        } finally {
            close(rs);
            close(stmt);
            close(conn);
        }
        return fieldNamesWithType;
    }


    /**
     * 获取表字段名、类型
     *
     * @param datasourceModel
     * @param tableName
     * @return
     * @throws SQLException
     */
    public static Map<String, String> showTableFieldsAndType(DatasourceModel datasourceModel, String tableName) {
        DataBaseType dataBaseType = getEnumByKey(datasourceModel.getType());
        Assert.notNull(dataBaseType, "DataX暂不支持改该数据库类型: " + datasourceModel.getType());

        Connection conn = getConnection(datasourceModel);

        if (Oracle.equals(dataBaseType)) {
            tableName = tableName.toUpperCase();
        }

        ResultSet rs = null;
        try {
            //获得元数据
            DatabaseMetaData metaData = conn.getMetaData();
            //通过表名获得所有字段名
            rs = metaData.getColumns(null, "%", tableName, "%");
        } catch (SQLException throwables) {
            throwables.printStackTrace();
            log.error("获取连接元数据失败！");
            throw RdbmsException.asCustomException("获取连接元数据失败");
        }

        //获得字段名
        String columnName = null;
        //字段类型
        String columnType = null;

        Map<String, String> columns = Maps.newHashMap();
        try {
            while (rs.next()) {
                columnName = rs.getString("COLUMN_NAME");

                switch (rs.getInt("DATA_TYPE")) {
                    case Types.CHAR:
                    case Types.NCHAR:
                    case Types.VARCHAR:
                    case Types.LONGVARCHAR:
                    case Types.NVARCHAR:
                    case Types.LONGNVARCHAR:
                        columnType = "string";
                        break;

                    case Types.CLOB:
                    case Types.BLOB:
                    case Types.NCLOB:

                    case Types.BINARY:
                    case Types.VARBINARY:
                    case Types.LONGVARBINARY:
                        columnType = "binary";
                        break;

                    case Types.SMALLINT:
                    case Types.TINYINT:
                    case Types.INTEGER:
                    case Types.BIGINT:
                        columnType = "int";
                        break;

                    case Types.BOOLEAN:
                        columnType = "boolean";
                        break;

                    case Types.BIT:
                        columnType = "bit";
                        break;

                    case Types.TIMESTAMP:
                    case Types.DATE:
                    case Types.TIME:
                        columnType = "time";
                        break;

                    case Types.NUMERIC:
                    case Types.FLOAT:
                    case Types.REAL:
                    case Types.DOUBLE:
                    case Types.DECIMAL:
                        columnType = "float";
                        break;

                    default:
                        throw CustomException.asCustomException(
                                DBUtilErrorCode.UNSUPPORTED_TYPE,
                                String.format(
                                        "您的配置文件中的列配置信息有误. 因为DataX 不支持数据库读取这种字段类型. 字段名:[%s], 字段类型:[%s]. 请尝试使用数据库函数将其转换datax支持的类型 或者不同步该字段 .",
                                        columnName, rs.getString("TYPE_NAME")));
                }
                columns.put(columnName, columnType);
            }

        } catch (SQLException throwables) {
            throwables.printStackTrace();
            log.error("解析数据库表字段信息失败！");
            throw CustomException.asCustomException(DBUtilErrorCode.UNSUPPORTED_TYPE, "解析数据库表字段信息失败！");
        } finally {
            JdbcCustomUtils.close(rs);
            JdbcCustomUtils.close(conn);
        }

        return columns;
    }

}
